package com.chwl.framework.grpc.etcd;

import com.alibaba.fastjson.JSON;
import com.chwl.framework.grpc.exception.UserdefinedException;
import io.etcd.jetcd.*;
import io.etcd.jetcd.common.exception.ClosedClientException;
import io.etcd.jetcd.kv.DeleteResponse;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.options.DeleteOption;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.options.WatchOption;
import io.etcd.jetcd.watch.WatchResponse;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


/**
 * 包名：com.mechat.connector.etcd<br>
 * <p>功能:etcd封装</p>
 *
 * @date:2018-07-12 下午6:56<br/>
 * @version:1.0 <br/>
 */
public class EtcdCrudService {
    protected static Logger heartbeatLogger = LoggerFactory.getLogger("heartbeat");
    private static Logger logger = LoggerFactory.getLogger(EtcdCrudService.class);
    private static EtcdCrudService instance = new EtcdCrudService();
    protected static Charset utf8 = Charset.forName("utf-8");
    protected ExecutorService executorService = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new DefaultThreadFactory("etcd-heartbeat-watch-service"));

    public EtcdCrudService() {
    }

    public static EtcdCrudService getInstance() {
        return instance;
    }

    Client client = null;
    Lease leaseClient = null;
    String[] endpoints;

    /**
     * 打开etcd连接
     *
     * @param endpoints
     * @throws UserdefinedException
     */
    public void connection(String... endpoints) throws UserdefinedException {
        if (endpoints == null || endpoints.length == 0) {
            throw new UserdefinedException("etcd client connection fail,endpoints is empty");
        }
        this.endpoints = endpoints;
        logger.info("etcd client connection start,connection servers:{}", JSON.toJSONString(endpoints));
        client = Client.builder().endpoints(endpoints).build();
        leaseClient = client.getLeaseClient();
        logger.info("etcd client connection successful");
    }

    public Client getClient() {
        return client;
    }

    public Lease getLeaseClient() {
        return leaseClient;
    }

    /**
     * 构建过期时间
     *
     * @return
     */
    public PutOption buildDefaultPutOption() throws Exception {
        return PutOption.newBuilder().build();

    }

    /**
     * 构建过期时间
     *
     * @param leaseId 租约编号
     * @return
     */
    public PutOption buildPutOptionByLeaseId(long leaseId) throws Exception {
        if (leaseId == -1) {
            return buildDefaultPutOption();
        }
        return PutOption.newBuilder().withLeaseId(leaseId).build();
    }

    /**
     * 获取租约id
     *
     * @param ttl
     * @return
     * @throws Exception
     */
    public long buildLeaseId(long ttl) throws Exception {
        if (ttl == -1) {
            return -1;
        }
        logger.info("etcd crud service,build leaseId,ttl:{}", ttl);
        return getLeaseClient().grant(ttl).get().getID();
    }

    /**
     * 写入
     *
     * @param key
     * @param value
     * @throws UserdefinedException
     */
    public void put(String key, String value) throws UserdefinedException {
        put(key, value, -1);
    }

    /**
     * 写入
     *
     * @param key
     * @param value
     * @param ttl   过期时间 单位为秒
     * @throws UserdefinedException
     */
    public long put(String key, String value, long ttl) throws UserdefinedException {
        try {
            long leaseId = buildLeaseId(ttl);
            put(key, value, buildPutOptionByLeaseId(leaseId));
            return leaseId;
        } catch (Exception e) {
            logger.error("etcd client,put data fail,key:{},value:{},ttl:{},error message:{}", key, value, ttl, e.getMessage(), e);
            throw new UserdefinedException("etcd client,put data fail");
        }

    }

    /**
     * 写入
     *
     * @param key
     * @param value
     * @param option 过期时间 单位为秒
     * @throws UserdefinedException
     */
    public void put(String key, String value, PutOption option) throws UserdefinedException {
        if (StringUtils.isEmpty(key)) {
            throw new UserdefinedException("key should not be empty");
        }
        if (StringUtils.isEmpty(value)) {
            throw new UserdefinedException("value should not be empty");
        }
        if (getClient() == null) {
            throw new UserdefinedException("etcd client is not connection");
        }
        try {
            getClient().getKVClient().put(ByteSequence.from(key, utf8), ByteSequence.from(value, utf8), option).get();
        } catch (Exception e) {
            logger.error("etcd client,put data fail,key:{},value:{},error message:{}", key, value, e.getMessage(), e);
            throw new UserdefinedException("etcd client,put data fail");
        }
    }

    /**
     * 获取单个
     *
     * @param key
     * @return
     * @throws UserdefinedException
     */
    public String get(String key) throws UserdefinedException {
        List<KeyValue> list = getList(key);
        if (CollectionUtils.isEmpty(list)) {
            return StringUtils.EMPTY;
        }
        return list.get(0).getValue().toString(utf8);
    }


    /**
     * 根据获取集合
     *
     * @param key
     * @return
     * @throws UserdefinedException
     */
    public List<KeyValue> getList(String key) throws UserdefinedException {
        if (StringUtils.isEmpty(key)) {
            logger.error("etcd client,get data is empty,key is empty");
            return Collections.emptyList();
        }
        if (getClient() == null) {
            throw new UserdefinedException("etcd client is not connection");
        }
        try {
            GetResponse response = getClient().getKVClient().get(ByteSequence.from(key, utf8), GetOption.newBuilder().build()).get();
            List<KeyValue> list = response.getKvs();
            if (CollectionUtils.isEmpty(list)) {
                logger.error("etcd client,get list data is empty,key:{}", key);
                return Collections.emptyList();
            }
            return response.getKvs();
        } catch (Exception e) {
            logger.error("etcd client,get list data fail,key:{},error message:{}", key, e.getMessage(), e);
        }
        return Collections.emptyList();
    }

    /**
     * 根据前缀获取集合
     *
     * @param prefixKey
     * @return
     * @throws UserdefinedException
     */
    public List<KeyValue> findList(String prefixKey) {
        try {
            GetOption.Builder builder = GetOption.newBuilder();
            builder.withPrefix(ByteSequence.from(prefixKey, utf8));
            GetOption option = builder.build();

            GetResponse response = getClient().getKVClient().get(ByteSequence.from(prefixKey, utf8), option).get();
            List<KeyValue> list = response.getKvs();
            if (CollectionUtils.isEmpty(list)) {
                logger.error("etcd client,find list data is empty,key:{}", prefixKey);
                return Collections.emptyList();
            }
            return response.getKvs();
        } catch (Exception e) {
            logger.error("etcd client,find list data fail,key:{},error message:{}", prefixKey, e.getMessage(), e);
        }
        return Collections.emptyList();
    }

    /**
     * 根据前缀删除
     *
     * @param prefixKey
     * @return
     * @throws UserdefinedException
     */
    public long delete(String prefixKey) {
        try {
            DeleteOption option = DeleteOption.newBuilder().withPrefix(ByteSequence.from(prefixKey, utf8)).build();
            DeleteResponse response = getClient().getKVClient().delete(ByteSequence.from(prefixKey, utf8), option).get();
            return response.getDeleted();
        } catch (Exception e) {
            logger.error("etcd client,delete fail,prefixKey:{},error message:{}", prefixKey, e.getMessage(), e);
        }
        return 0;
    }

    /**
     * 根据前缀删除
     *
     * @param key
     * @return
     * @throws UserdefinedException
     */
    public long deleteByKey(String key) {
        try {
            if (StringUtils.isEmpty(key)) {
                logger.error("etcd client,delete fail,key is not empty");
                return 0;
            }
            DeleteResponse response = getClient().getKVClient().delete(ByteSequence.from(key, utf8)).get();
            return response.getDeleted();
        } catch (Exception e) {
            logger.error("etcd client,delete fail,key:{},error message:{}", key, e.getMessage(), e);
        }
        return 0;
    }


    /**
     * watch监听
     *
     * @param key
     * @param listener
     * @throws Exception
     */
    public void watch(String key, EtcdWatchListener listener) {
        Watch.Watcher watcher = null;
        try {
            ByteSequence keyByte = ByteSequence.from(key, utf8);
            //监听前缀为X的节点
            WatchOption watchOption = WatchOption.newBuilder().withPrefix(keyByte).build();
            watcher = getClient().getWatchClient().watch(keyByte, watchOption, new Watch.Listener() {
                @Override
                public void onNext(WatchResponse response) {
                    listener.listener(response.getEvents());
                }

                @Override
                public void onError(Throwable throwable) {

                }

                @Override
                public void onCompleted() {

                }
            });
            //            while (true) {
            //                WatchResponse response = watcher.listen();
            //                listener.listener(response.getEvents());
            //            }
        } catch (Exception e) {
            if (watcher != null) {
                watcher.close();
            }
            if (!(e instanceof ClosedClientException) && !(e instanceof InterruptedException)) {
                logger.error("etcd client watch fail,error message:{}", e.getMessage(), e);
            }
        }
    }


    /**
     * 心跳上报
     *
     * @param listener
     * @param heartbeatTime
     */
    public void heartbeat(HeartbeatListener listener, int heartbeatTime) {
        logger.info("etcd client heartbeat start,heartbeatTime:{}", heartbeatTime);
        executorService.execute(() -> {
            while (true) {
                try {
                    //heartbeatTime秒*2倍时间过期
                    listener.heartbeat(getClient().getKVClient(), buildDefaultPutOption());
                    Thread.sleep(heartbeatTime * 1000);
                } catch (Exception e) {
                    logger.error("etcd client heartbeat fail,heartbeatTime:{},error message:{}", heartbeatTime * 1000, e.getMessage(), e);
                }
            }
        });
    }

    public interface HeartbeatListener {
        void heartbeat(KV kv, PutOption option);

    }

    /**
     * 销毁连接
     *
     * @return
     * @throws UserdefinedException
     */
    public void destroy() {
        try {
            if (getClient() != null) {
                getClient().close();
                logger.info("etcd client,destroy successful");
            }
            if (getLeaseClient() != null) {
                getLeaseClient().close();
                logger.info("etcd lease client,destroy successful");
            }
            logger.info("etcd connector close successful");
        } catch (ClosedClientException e) {
            logger.error("etcd lease client,destroy fail,error message:{}", e.getMessage());
        } catch (Throwable ex) {
            logger.error("etcd lease client,destroy fail,error message:{}", ex.getMessage(), ex);
        }
    }
}